

package org.apache.flink.api.common.eventtime;

import org.apache.flink.annotation.Public;

import java.time.Duration;

import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;

/*
        JVM的类加载的双亲委派模型。
            如果有两个同名的类，一个是自己写的，一个是引入别人的。
                优先加载你自己写的
 */
@Public
public class BoundedOutOfOrdernessWatermarks<T> implements WatermarkGenerator<T> {

    /** The maximum timestamp encountered so far. */
    private long maxTimestamp;

    /** The maximum out-of-orderness that this watermark generator assumes. */
    private final long outOfOrdernessMillis;

    /**
     * Creates a new watermark generator with the given out-of-orderness bound.
     *
     * @param maxOutOfOrderness The bound for the out-of-orderness of the event timestamps.
     */
    public BoundedOutOfOrdernessWatermarks(Duration maxOutOfOrderness) {
        checkNotNull(maxOutOfOrderness, "maxOutOfOrderness");
        checkArgument(!maxOutOfOrderness.isNegative(), "maxOutOfOrderness cannot be negative");

        this.outOfOrdernessMillis = maxOutOfOrderness.toMillis();

        // start so that our lowest watermark would be Long.MIN_VALUE.
        this.maxTimestamp = Long.MIN_VALUE + outOfOrdernessMillis + 1;
    }

    // ------------------------------------------------------------------------

    //基于数据，生成水印
    // WatermarkOutput output 发水印
    // 来一条数据才调用一次
    // 间歇性发送，来一条数据，发一次
    @Override
    public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
        //maxTimestamp 指eventTime中最大的时间属性
        maxTimestamp = Math.max(maxTimestamp, eventTimestamp);

        //Watermark watermark = new Watermark(maxTimestamp - outOfOrdernessMillis - 1);
        //System.out.println("收到了数据:"+event);
        //System.out.println("向下游发送了:"+watermark.getTimestamp());
        //output.emitWatermark(watermark);
    }

    //周期型(默认200ms)发送水印
    // outOfOrdernessMillis: 乱序容忍度，如果调用的是 forMonotonousTimestamps()，代表不是乱序，此时 outOfOrdernessMillis = 0
    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        Watermark watermark = new Watermark(maxTimestamp - outOfOrdernessMillis - 1);
        //System.out.println(Thread.currentThread().getName()+"向下游发送了:"+watermark.getTimestamp());
        output.emitWatermark(watermark);
    }
}
